package com.cjz.demo8;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

/**
 * Flowable介绍
 * Created by cjz on 2016/12/1.
 */

public class Test {
    public static void main(String[] args) throws InterruptedException {

        testDemo1();
//        testDemo2();

        Thread.sleep(15_000);
    }

    private static void testDemo1() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> e) throws Exception {
                for (int i = 0; i < 1000; i++) {
                    e.onNext(i);
                }
                e.onComplete();
            }
        }, BackpressureStrategy.BUFFER)
                .subscribeOn(Schedulers.computation())
                .observeOn(Schedulers.newThread())
                .subscribe(new Consumer<Integer>() {
                               @Override
                               public void accept(Integer integer) throws Exception {
                                   System.out.println(integer.toString());
                                   Thread.sleep(10);
                               }
                           },
                        new Consumer<Throwable>() {
                            @Override
                            public void accept(Throwable throwable) throws Exception {
                                System.out.println(throwable.getMessage());
                            }
                        },
                        new Action() {
                            @Override
                            public void run() throws Exception {
                                System.out.println("complete");
                            }
                        }
                );
    }

    private static void testDemo2() {
        Flowable.<Integer>create(
                e -> {
                    for (int i = 0; i < 1000; i++) {
                        e.onNext(i);
                    }
                    e.onComplete();
                },
                BackpressureStrategy.BUFFER
        )
                .subscribeOn(Schedulers.computation())
                .observeOn(Schedulers.newThread())
                .subscribe(
                        integer -> {
                            System.out.println(integer.toString());
                            Thread.sleep(10);
                        },
                        throwable -> System.out.println(throwable.getMessage()),
                        () -> System.out.println("complete")
                );
    }

}
